package solution.sort;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.PriorityQueue;

import org.apache.commons.lang.StringUtils;
import org.junit.Test;

/**
 * @author <a href="www.sureinterview.com">SureInterview</a>
 */
public class MergeNSorted {
	/**
	 * The interface that can be adapted to any device, such as file.
	 */
	public interface AStream {
		Integer curData();

		Integer readNext();
	}

	/**
	 * simple implementation of ASteam.
	 */
	class ListStream implements AStream {
		List<Integer> dataList;

		int pos = 0;
		int size = 0;

		public ListStream(List<Integer> data) {
			dataList = data;
			size = dataList.size();
		}

		/**
		 * when all data are calculated, return the SUPER_LARGE marker.
		 */
		@Override
		public Integer curData() {
			if (pos >= size)
				return SUPER_LARGE;
			return dataList.get(pos);
		}

		/**
		 * get next piece of data.
		 */
		@Override
		public Integer readNext() {
			if (pos >= size)
				return SUPER_LARGE;
			return dataList.get(pos++);
		}

		@Override
		public String toString() {
			return "[" + StringUtils.join(dataList, ",") + "]";
		}
	}

	/**
	 * holds stream data.
	 */
	class QElem implements Comparable<QElem> {
		AStream stream;

		public QElem(AStream stream) {
			this.stream = stream;
		}

		/**
		 * Tell the priority queue how to compare.
		 */
		@Override
		public int compareTo(QElem qElem) {
			Integer curData = stream.curData();
			Integer qElmData = qElem.stream.curData();
			if (qElmData == SUPER_LARGE && curData == SUPER_LARGE)
				return 0;
			if (curData == SUPER_LARGE)
				return 1;
			if (qElmData == SUPER_LARGE)
				return -1;
			return curData.compareTo(qElmData);
		}
	}

	/**
	 * Marker of the end, which is defined as to be larger than any number.
	 */
	final Integer SUPER_LARGE = null;

	/**
	 * Merge the sorted data stream into one large sorted data.
	 * 
	 * @param dataLists
	 * @return
	 */
	List<Integer> merge(AStream[] dataLists) {
		if (dataLists == null || dataLists.length == 0)
			return null;

		// 1. prepare the queue. Build the heap for first element of the queue.
		// Note that the QElem is comparable for the first element.
		PriorityQueue<QElem> priQ = new PriorityQueue<QElem>();
		for (AStream data : dataLists) {
			if (data == null) {
				continue;
			}
			QElem qElem = new QElem(data);
			priQ.add(qElem);
		}

		// 2. heap is ready. Pull out the data one by one from the priority
		// queue. SUPER_LARGE is the marker of end of stream.
		List<Integer> merged = new ArrayList<Integer>();
		do {
			QElem qElem = priQ.poll();
			Integer data = qElem.stream.curData();
			if (data == SUPER_LARGE) {
				continue;
			}
			// still a valid data, put the element back.
			merged.add(data);
			qElem.stream.readNext();
			priQ.add(qElem);
		} while (!priQ.isEmpty());

		return merged;
	}

	/**
	 * merge 2 sorted array.
	 * 
	 * @param data1
	 * @param data2
	 * @return the merged array
	 */
	Integer[] merge2(Integer[] data1, Integer[] data2) {
		// 1. no need to merge when one queue is empty
		if (data1 == null)
			return data2;
		if (data2 == null)
			return data1;

		// 2. merge
		int p1 = 0, p2 = 0, m = 0;
		Integer[] mrgData = new Integer[data1.length + data2.length];
		while (p1 < data1.length && p2 < data2.length) {
			if (data1[p1] < data2[p2]) {
				mrgData[m++] = data1[p1++];
			} else {
				mrgData[m++] = data2[p2++];
			}
		}

		// 3. handing remaining data still in the queue.
		while (p1 < data1.length) {
			mrgData[m++] = data1[p1++];
		}
		while (p2 < data2.length) {
			mrgData[m++] = data2[p2++];
		}
		return mrgData;
	}

	@Test
	public void test() {
		Integer[][] data = new Integer[][] {//
		// test data
				{ 1, 2, 3, 4, 5, 6, 7, 8 },//
				{ 3, 5, 7, 9, 11 },//
				{ 8, 8, 8, 8 },//
				{ 2, 6, 10, 14 },//
		};

		// prepare data
		AStream[] stream = new AStream[data.length];
		for (int i = 0; i < data.length; i++) {
			stream[i] = new ListStream(Arrays.asList(data[i]));
		}

		// merge
		List<Integer> mgdata = merge(stream);
		System.out.println(StringUtils.join(mgdata, ","));
	}

	@Test
	public void test2() {
		Integer[][] data = new Integer[][] {//
		// test data
				{ 1, 2, 3, 4, 5, 6, 7, 8 },//
				{ 3, 5, 7, 9, 11 },//
				{ 8, 8, 8, 8 },//
				{ 2, 6, 10, 14 },//
		};

		// prepare data
		for (int i = 0; i < data.length; i += 2) {
			Integer[] mrgData = merge2(data[i], data[i + 1]);
			System.out.println(StringUtils.join(mrgData, ","));
		}
	}
}
